package com.tech

import com.tech.common.KuduUtil.getIdsByGroupId
import com.tech.common.{KafkaUtil, KuduUtil, MysqlUtil, SimplyDateUtil}
import com.tech.config.ApplicationConfig
import com.tech.process.nodeProcess.{BehaviorNode, NextNode, TestNode, WaitNode}
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods

import java.sql.{Connection, DriverManager}
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
import scala.util.parsing.json.JSON


//156430d5-67af-4475-b7a6-1f5140f8d301

/**
 * 初始化群组活动
 */
object InitializeGroupActivity {

  def main(args: Array[String]): Unit = {

    println("System.currentTimeMillis():" + System.currentTimeMillis())

    val connection: Connection = DriverManager.getConnection(ApplicationConfig.URL_MYSQL)

    //租户id
    val project = ApplicationConfig.PROJECT
    //活动id
    val activityId = args(1)

    println(activityId)
    //获取活动开始节点map（活动id -> 开始id）
    val startMap = MysqlUtil.getStartNodes(connection, ArrayBuffer(activityId))
    //获取下级节点map（节点id -> 下级节点NextNode）
    val nextNodeMap = MysqlUtil.getNextNodes(connection, ArrayBuffer(activityId))
    //获取客群map（客群id -> set(客户id)）
    val groupMap = MysqlUtil.getGroups(connection, ArrayBuffer(activityId))
    //获取所有客户节点(静态)
    val groupNodeMap = MysqlUtil.getGroupNodesWithOutDynamic(connection, activityId)
    //获取A/B Test节点map
    val testNodeMap = MysqlUtil.getTestNodes(connection, ArrayBuffer(activityId))
    //获取行为节点map
    val behaviorNodeMap = MysqlUtil.getBehaviorNodes(connection, ArrayBuffer(activityId))
    //获取等待节点map
    val waitNodeMap = MysqlUtil.getWaitNodes(connection, ArrayBuffer(activityId))


    //初始化客群表数据
    KuduUtil.insertGroups4Map(groupMap, project, activityId)
    //依次处理所有客户节点
    groupNodeMap.foreach(
      groupNode => {

        println("groupId:" + groupNode._2)
        //保存客户参与活动记录
        KuduUtil.insertCustomerActivity4Group(groupNode._2, activityId)
        KafkaUtil.send2Kafka4Group(ApplicationConfig.CUSTOMER_NODE_TOPIC, project, activityId, groupNode._1, groupNode._2)
        passNode(project, groupNode._2, activityId, groupNode._1, behaviorNodeMap, nextNodeMap, testNodeMap, waitNodeMap, groupMap)

      }
    )

    println("执行完毕")


  }

  def passNode(project: String,
               groupId: String,
               activity_id: String,
               node: String,
               behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
               nextNodeMap: mutable.HashMap[String, NextNode],
               testNodeMap: mutable.HashMap[String, TestNode],
               waitNodeMap: mutable.HashMap[String, WaitNode],
               groupMap: mutable.HashMap[String, mutable.Set[String]]): Unit = {

    def toNextNode(currentNode: String, currentGroupId: String): Unit = {
      val nextIds = nextNodeMap(currentNode).next_id

      if (nextIds != null)
        nextIds.split(",").foreach(
          next =>
            //100000000 开始
            //100000001 结束
            //100000002 用户
            //100000003 A/B Test
            //100000004 A/B Test Child
            //100000005 行为
            //100000006 奖励
            //100000007 等待
            //100000008 消息
            nextNodeMap(next).id_type match {
              case 100000000 =>
              case 100000001 =>
              case 100000002 =>
              case 100000003 =>

                KafkaUtil.send2Kafka4Group(ApplicationConfig.CUSTOMER_NODE_TOPIC, project, activity_id, next, currentGroupId)
                //客群
                val customers = groupMap(currentGroupId)
                //A/B Test分组array
                val array = ArrayBuffer[String]()
                val map = mutable.HashMap[String, mutable.Set[String]]()
                //客群大小
                val groupSize = customers.size
                println("groupSize：" + groupSize)
                //各个子分支承载量
                val pros = mutable.HashMap[String, Double]()
                nextNodeMap(next).next_id.split(",").foreach(
                  tmpNode => {
                    println(testNodeMap(tmpNode))
                    pros.put(tmpNode, testNodeMap(tmpNode).probability.toDouble / 100 * groupSize)
                    array.append(tmpNode)
                  }
                )
                customers.foreach(
                  customer => {
                    if (array.nonEmpty) {
                      val index = new Random().nextInt(array.length)
                      val newGroupId = array(index)
                      if (map.contains(newGroupId)) {
                        map(newGroupId).add(customer)
                      } else {
                        val tmpSet = mutable.Set[String](customer)
                        map.put(newGroupId, tmpSet)
                      }
                      if (map(newGroupId).size >= pros(newGroupId)) {
                        println(map(newGroupId).size, "     ", pros(newGroupId))
                        array.remove(index)
                      }
                    }
                  }
                )
                println(map.size)
                println(map.foreach(t => println(t._1, "         ", t._2.size)))

                //将map中的数据写入到kudu的客群表中
                println(System.currentTimeMillis())
                map.foreach(
                  i => {
                    groupMap.put(i._1, i._2)
                  }
                )
                KuduUtil.insertGroups4Map(map, project, activity_id)
                println(System.currentTimeMillis())

                pros.foreach(
                  i => {
                    //客户群组表已改变
                    println("i._1:" + i._1)
                    KafkaUtil.send2Kafka4Group(ApplicationConfig.CUSTOMER_NODE_TOPIC, project, activity_id, i._1, i._1)
                    toNextNode(i._1, i._1)
                  }
                )

              case 100000004 =>
                KafkaUtil.send2Kafka4Group(ApplicationConfig.CUSTOMER_NODE_TOPIC, project, activity_id, next, currentGroupId)
                //直接通过
                toNextNode(next, currentGroupId)
              case 100000005 =>
                //实现一个将map（String -> Set）批量写入kudu的方法
                println("开始写入customer_node表")
                println(currentGroupId)
                println(getIdsByGroupId(currentGroupId))

                //是否存在累计满足条件
                if (behaviorNodeMap(next).addition != null && behaviorNodeMap(next).addition.contains("100000000")) {

                  val json1Obj = JSON.parseFull(behaviorNodeMap(next).measure_property)
                  val map1 = json1Obj.get.asInstanceOf[Map[String, Any]]

                  val limitSymbol = map1("limitSymbol").asInstanceOf[String]
                  val limitName = map1("limitName").asInstanceOf[List[String]]
                  if (limitSymbol.equals("between")) {
                    val limitValue = map1("limitValue").asInstanceOf[String]
                    val minValue = limitValue.split(",").head.toDouble
                    val maxValue = limitValue.split(",")(1).toDouble
                    if (limitName.length == 1) {
                      KuduUtil.upsertNode4Map(currentGroupId, activity_id, next, behaviorNodeMap(next).event_type, is_delete = null,
                        System.currentTimeMillis(), limitName.head, limitSymbol, limit_value_min = minValue, limit_value_max = maxValue, event_value = 0, event_num = 0)
                    } else {
                      KuduUtil.upsertNode4Map(currentGroupId, activity_id, next, behaviorNodeMap(next).event_type, is_delete = null,
                        System.currentTimeMillis(), limitName.head + "|" + limitName(1), limitSymbol, limit_value_min = minValue, limit_value_max = maxValue, event_value = 0, event_num = 0)
                    }
                  } else {
                    val limitValue = map1("limitValue").asInstanceOf[Double]
                    if (limitName.length == 1) {
                      KuduUtil.upsertNode4Map(currentGroupId, activity_id, next, behaviorNodeMap(next).event_type, is_delete = null,
                        System.currentTimeMillis(), limitName.head, limitSymbol, limit_value = limitValue, event_value = 0, event_num = 0)
                    } else {
                      KuduUtil.upsertNode4Map(currentGroupId, activity_id, next, behaviorNodeMap(next).event_type, is_delete = null,
                        System.currentTimeMillis(), limitName.head + "|" + limitName(1), limitSymbol, limit_value = limitValue, event_value = 0, event_num = 0)
                    }
                  }
                } else {
                  KuduUtil.upsertNode4Map(currentGroupId, activity_id, next, behaviorNodeMap(next).event_type, is_delete = null,
                    System.currentTimeMillis(), event_value = 0, event_num = 0)
                }

                println("结束写入customer_node表")

                implicit val f: DefaultFormats.type = org.json4s.DefaultFormats
                val timeMap = JsonMethods.parse(behaviorNodeMap(next).time_condition).extract[Map[String, Any]]
                val action_time = if (timeMap.contains("endTime")) {
                  SimplyDateUtil.getFormat().parse(timeMap("endTime").toString).getTime
                } else {
                  val count = timeMap("count").toString.toDouble.toInt
                  //100000000年 100000001月 100000002周 100000003天
                  System.currentTimeMillis() + (timeMap("unitTime").toString.toDouble.toInt match {
                    case 100000000 => count.toLong * 1000 * 60 * 60 * 24 * 365
                    case 100000001 => count.toLong * 1000 * 60 * 60 * 24 * 30
                    case 100000002 => count.toLong * 1000 * 60 * 60 * 24 * 7
                    case 100000003 => count.toLong * 1000 * 60 * 60 * 24
                  })
                }

                println("action_time:" + action_time)

                //判断是否节点包含累计满足的规则
                if (behaviorNodeMap(next).addition != null && behaviorNodeMap(next).addition.contains("100000000")) {

                  val mpMap = JsonMethods.parse(behaviorNodeMap(next).measure_property).extract[Map[String, Any]]

                  println(mpMap("limitName").getClass.getSimpleName)

                  val limitSymbol = mpMap("limitSymbol").toString
                  if (limitSymbol != "between") {
                    val limitValue = mpMap("limitValue").toString.toDouble
                    KuduUtil.insertDelayNode4Map(currentGroupId, project, activity_id, next, action_time, "behavior", behaviorNodeMap(next).event_type, current_value = 0, limit_value = limitValue, limit_symbol = limitSymbol)
                  } else {
                    val limitValue = mpMap("limitValue").toString
                    val minLimitValue = limitValue.split(",").head.toDouble
                    val maxLimitValue = limitValue.split(",")(1).toDouble
                    KuduUtil.insertDelayNode4Map(currentGroupId, project, activity_id, next, action_time, "behavior", behaviorNodeMap(next).event_type, current_value = 0, limit_value_min = minLimitValue, limit_value_max = maxLimitValue, limit_symbol = limitSymbol)
                  }
                } else {
                  KuduUtil.insertDelayNode4Map(currentGroupId, project, activity_id, next, action_time, "behavior", behaviorNodeMap(next).event_type)
                }
              case 100000006 | 100000008 =>
                KafkaUtil.send2Kafka4Group(ApplicationConfig.CUSTOMER_NODE_TOPIC, project, activity_id, next, currentGroupId)
                toNextNode(next, currentGroupId)
              case 100000007 =>
                // 发送消息到延时事件表
                val action_time = if (waitNodeMap(next).wait_type == 100000000) {
                  System.currentTimeMillis() + (waitNodeMap(next).unit match {
                    case 100000001 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24 * 30
                    case 100000002 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24 * 7
                    case 100000003 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24
                    case 100000004 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60
                  })
                } else {
                  waitNodeMap(next).wait_date * 1000
                }

                println("action_time:" + action_time)
                KuduUtil.insertDelayNode(project, "", "", activity_id, next, action_time, "wait", is_group = "1", group_id = next)

              case _ =>
                println("未识别的类型：" + nextNodeMap(next).id_type)

            }
        )

    }

    toNextNode(node, groupId)

  }


}
